package com.bw.gmall.app.DIM;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.bean.TableProcess;
import com.bw.gmall.utils.MyKafkaUtil;
import com.bw.gmall.utils.MysqlUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/*
1.获取topic_db数据
2.获取flinkcdc配置表数据
3. 动态创建配置表  维度表
4. 拿着配置表数据  和topic_db进行匹配
5. 匹配上  就把该条数据插入 维度表中
*
* */
public class DimApp {
    public static void main(String[] args) throws Exception {
//1.获取环境遍历
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//2.设置并行度
        env.setParallelism(1);
//设置检查点与状态后端
//开启Checkpoint
//         env.enableCheckpointing(5*60000L, CheckpointingMode.EXACTLY_ONCE);
 //超时  checkpoint 最多只能10分钟完成
//        env.getCheckpointConfig().setCheckpointTimeout(10*60000L);
//共存状态     比如我一个保存点 8分钟完成   理论上完的第二个checkpoint应该是在5分钟开启
//如果是1  只能8分钟之后完成  ,但是现在是2 说明可以在5分钟之后完成   可以共存状态
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
//如果我约到问题了
// 一个时间段内的最大失败次数
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000L));
//设置状态后端  把状态存放在内存里
//每传入一条数据,有状态的算子任务都会读取和更新状态。
//由于有效的状态访问对于处理数据的低延迟至关重要,因此每个并行任务子任务都会在本地维护其状态
// 以确保快速的状态访状态的存储、访问以及维护,由一个可插入的组件决定,这个组件就叫做状态后端
//        env.setStateBackend(new HashMapStateBackend());
//储配置存储检查点到文件系统，可以直接传入一个 String，指定文件系统的路径
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/211126/ck");
//如果在本地运行需要紧加上这句话
//        System.setProperty("HADOOP_USER_NAME","ROOT");
//2.获取topic_db数据
        String topic="topic_db";
        String groupId="dimapp";
        DataStreamSource<String> source = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
 source.print("topic_db数据>>>>>>>");
//3.过滤非json 数据   保留新增及变化及初始化数据
        SingleOutputStreamOperator<JSONObject> filterdb = source.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(value);
                    String type = jsonObject.getString("type");
                    if ("insert".equals(type) || "update".equals(type) || "bootstrap-insert".equals(type)) {
                        out.collect(jsonObject);
                    }
                } catch (Exception e) {
                    System.out.println("数据错误，不是json格式" + value);

                }
            }
        });
        filterdb.print();

//TODO 4. 使用FlinkCDC  读取mysql 配置信息表    创建配置流
//维度表
        DataStream<String> cdcdb = MysqlUtil.cdcMysql(env, "gmall_config", "table_process");
//        cdcdb.print()   ;
//todo 5. 将配置流转换为广播流
        
        // 创建一个MapStateDescriptor，用于描述广播状态
        // 状态名称为"mapState"，key为String类型（源表名），value为TableProcess类型（配置信息）
        MapStateDescriptor<String, TableProcess> mapState
                = new MapStateDescriptor<>("mapState", String.class, TableProcess.class);
        //广播流
        BroadcastStream<String> broadcast = cdcdb.broadcast(mapState);

        //连接主流和广播流
        BroadcastConnectedStream<JSONObject, String> connect = filterdb.connect(broadcast);
        //处理连接流，   根据配置处理主流数据（将配置信息存入到状态中  主流读状态）


        env.execute();
    }
}
